package com.edata.bigdata.flink;

import org.apache.flink.api.common.serialization.Encoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.OutputStream;
import java.io.Serializable;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;

public class CsvSinkEncoder<T> implements Encoder<T>, Serializable {
    public Logger logger = LoggerFactory.getLogger(this.getClass());
    private final Class<T> clazz;
    private boolean headerWritten = false;


    @Override
    public void encode(T t, OutputStream outputStream) throws IOException {
        try {
            Field[] fields = clazz.getDeclaredFields();
            int fieldCount = fields.length;
            //第一次写入：把列名先写入
            if (!headerWritten) {
                headerWritten = true;
                String[] fieldNames = new String[fieldCount];
                for (int i = 0; i < fields.length; i++) {
                    fieldNames[i] = fields[i].getName();
                }
                String headerLine = String.join(",", fieldNames) + "\n";
                logger.info("写入列名：{}", headerLine);
                outputStream.write(headerLine.getBytes(StandardCharsets.UTF_8));
            }
            String[] values = new String[fieldCount];
            for (int i = 0; i < fieldCount; i++) {
                fields[i].setAccessible(true);
                values[i] = fields[i].get(t).toString();
            }
            String line = String.join(",", values) + "\n";
            outputStream.write(line.getBytes(StandardCharsets.UTF_8));
        } catch (Exception e) {
            logger.error("无法写入数据：{}", e.getMessage());
        }
    }

    public CsvSinkEncoder(Class<T> clazz) {
        this.clazz = clazz;
    }
}
